package com.java.springcloud.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.UUID;

/**
 * @author dushougudu
 */
@Slf4j
@Service
public class RocketMqService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void send(String msg, String topic) {
        rocketMQTemplate.send(topic,
                MessageBuilder.withPayload(msg).build());
    }

    public void syncSendOrderly(String msg, String topic, String hashkey) {
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, MessageBuilder.withPayload(msg).build(), hashkey);
        log.info("队列消息发送结果：{}", sendResult.toString());
    }

    public void asyncSend(String msg, String topic) {
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(msg).build(),
                new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.info("消息发送成功：{}", sendResult.toString());
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        log.info("消息发送失败：{}", throwable.getMessage());
                    }
                });
    }

    public void sendIntransaction(String msg, String topic) {
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                .setHeader("tag", "mongodb")
                .build();
        TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(topic, message, topic);
        log.info("事务消息发送结果：{}", sendResult.toString());
    }


    public void sendDelay(String msg, String topic) {
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                .setHeader("tag", "mongodb")
                .build();
        SendResult sendResult = rocketMQTemplate.syncSend(topic, message, 10000L, 2);
        log.info("延迟消息发送结果：{}", sendResult.toString());
    }

}
